package org.piax.rospubsub;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

public class PublishRequestListener {
    private ADBBridge adbBridge = null;
    private InetSocketAddress endpoint = null;
    private int bufSize = -1;

    private ServerSocketChannel ssoc = null;
    private volatile boolean threadLoop = false;
    private Thread acceptThread = null;

    /**
     * コンストラクタ
     * @param adb_bridge ADB forward を使った Android との通信路
     * @param publishctrlport publish 要求受け付けポート番号
     * @param buf_size バッファサイズ
     */
    public PublishRequestListener(ADBBridge adb_bridge, int publishctrlport, int buf_size) {
        if (adb_bridge == null)
            throw new IllegalArgumentException("adb_bridge should not be null");

        adbBridge = adb_bridge;
        endpoint = new InetSocketAddress(publishctrlport);
        bufSize = buf_size;
    }

    /**
     * PublishProxy の待ち受けを開始する。
     * 
     * @throws IOException
     */
    public synchronized void start() throws IOException {
        if (threadLoop)
            throw new IllegalStateException("PublishProxy is already started");

        // クライアントからの要求を受けるソケットを開く
        ssoc = ServerSocketChannel.open();
        ssoc.socket().bind(endpoint);

        threadLoop = true;
        acceptThread  = new AcceptThread();
        acceptThread.start();
    }

    /**
     * PublishProxy の待ち受けを停止する。
     * 
     * @throws IOException
     * @throws InterruptedException 
     */
    public synchronized void stop() throws IOException, InterruptedException {
        if (!threadLoop)
            return;

        threadLoop = false;
        try {
            ssoc.close();
            acceptThread.interrupt();
            acceptThread.join();
        } finally {
            ssoc = null;
            acceptThread = null;
        }
    }

    /**
     *  接続受け付けスレッド
     */
    private class AcceptThread extends Thread {
        @Override
        public void run() {
            while (threadLoop) {
                try {
                    // クライアントからの接続を受け付け、受信スレッドを起こす
                    RecvThread t = new RecvThread(ssoc.accept());
                    t.start();
                } catch (IOException e) {
                    if (threadLoop) {
                        // threadLoop == true で IOException は切断などの例外事項
                        // 例外表示してループを抜ける
                        e.printStackTrace();
                        threadLoop = false;
                    }
                    // threadLoop == false での IOException は正常停止時
                }
            }
        }
    }

    /**
     * トピック名受信スレッド
     */
    private class RecvThread extends Thread {
        private SocketChannel soc = null;
        public RecvThread(SocketChannel soc) {
            if (soc == null)
                throw new IllegalArgumentException("soc should not be null");
            if (!soc.isConnected())
                throw new IllegalArgumentException("soc should be connected");

            this.soc = soc;
        }

        @Override
        public void run() {
            try {
                ByteBuffer buf = ByteBuffer.allocate(1024);

                soc.read(buf);  // トピック名受信
                buf.flip();
                byte[] topic_bytes = new byte[buf.remaining()];
                buf.get(topic_bytes);
                String topic = new String(topic_bytes, "UTF-8");

                PublishProxy p = new PublishProxy(topic.trim(), adbBridge, bufSize);

                buf.clear();
                buf.putInt(p.getLocalPort());
                buf.flip();
                soc.write(buf);     // トピックに対応するポート番号を応答
                soc.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
